package cn.dfun.sample.flink.tabletest
import cn.dfun.sample.flink.apitest.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.table.api.{EnvironmentSettings, Over, Table, Tumble}
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.types.Row

object TimeWindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 处理时间语义
//    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tableEnv = StreamTableEnvironment.create(env, settings)
    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    val inputStream= env.readTextFile(inputPath)
    //    val inputStream = env.socketTextStream("node-01", 7777)

    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
      })
    // 基于流创建表
    // 处理时间字段定义
    // 也可以在创建表的ddl中定义pt字段
//        val sensorTable = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'pt.proctime)
    // timestamp被转换为毫秒,timestamp为flink sql关键字定义别名
    val sensorTable = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts)

    // 1.group window
    val resultTable = sensorTable
      .window(Tumble over 10.seconds on 'ts as 'tw) // 每10秒统计一次滚动时间窗口 单一参数传递可以省略.和()
      .groupBy('id, 'tw)
      // 当前窗口结束时间
      .select('id, 'id.count, 'temperature.avg, 'tw.end)

    // 2 sql
    tableEnv.createTemporaryView("sensor", sensorTable)
    val resultSqlTable = tableEnv.sqlQuery(
      """
        |select id, count(id), avg(temperature), tumble_end(ts, interval '10' second)
        |from sensor
        |group by id, tumble(ts, interval '10' second)
      """.stripMargin)

    // overwindow
    // 统计每个数据与之前两行数据的平均温度
    val overResultTable = sensorTable
        .window(Over partitionBy 'id orderBy 'ts preceding 2.rows as 'ow)
        .select('id, 'ts, 'id.count over 'ow, 'temperature.avg over 'ow)

    // sql
    val overResultSqlTable = tableEnv.sqlQuery(
      """
        | select id, ts, count(id) over ow, avg(temperature) over ow
        | from sensor
        | window ow as (
        |   partition by id
        |   order by ts
        |   rows between 2 preceding and current row
        | )
      """.stripMargin)
    overResultTable.toAppendStream[Row].print("overresult")
    overResultSqlTable.toAppendStream[Row].print("oversqlresult")

//    resultTable.toAppendStream[Row].print("result")
//    resultSqlTable.toRetractStream[Row].print("sql")
//    sensorTable.printSchema()
//    sensorTable.toAppendStream[Row].print()
    env.execute("time window test")
  }
}
